package cn.datawin.cmppserver.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.nio.charset.StandardCharsets;

import static cn.datawin.constant.RocketMqConstant.CMPP_INPUT_TOPIC;

@Slf4j
@Configuration
public class RocketMQConfig {

    private final Environment environment;

    public RocketMQConfig(Environment environment) {
        this.environment = environment;
    }

    /**
     * 配置生产者
     */
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQProducer defaultMQProducer() {
        String nameServer = environment.getProperty("spring.rocketmq.name-server");
        String producerGroup = environment.getProperty("spring.rocketmq.producer.group");

        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(nameServer);
        producer.setVipChannelEnabled(false);

        log.info("RocketMQ Producer 初始化完成，NameServer: {}", nameServer);
        return producer;
    }

    /**
     * 配置消费者
     */
//    @Bean(initMethod = "start", destroyMethod = "shutdown")
//    public DefaultMQPushConsumer defaultMQPushConsumer() throws Exception {
//        String nameServer = environment.getProperty("spring.rocketmq.name-server");
//        String consumerGroup = environment.getProperty("spring.rocketmq.consumer.group");
//
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
//		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//		consumer.setNamesrvAddr(nameServer);
//
//        // 订阅主题和标签
//        consumer.subscribe(CMPP_INPUT_TOPIC, "*");
//
//        // 设置消息监听
//        consumer.registerMessageListener((MessageListenerConcurrently)
//                (msgs, context) -> {
//            for (MessageExt msg : msgs) {
//                try {
//                    String message = new String(msg.getBody(), StandardCharsets.UTF_8);
//					log.info("接收到消息: {}", message);
//                    // 处理消息业务逻辑
//                } catch (Exception e) {
//					log.error("处理消息失败", e);
//                    // 消费失败，返回重试
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                }
//            }
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        });
//
//		log.info("RocketMQ Consumer 初始化完成，NameServer: {}", nameServer);
//        return consumer;
//    }
}
